package memory

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"

	"github.com/louloulin/dataflare/pkg/core"
	"github.com/louloulin/dataflare/pkg/messages"
)

// MemorySourceConnector 是一个内存源连接器
type MemorySourceConnector struct {
	// data 是要读取的数据
	data []map[string]interface{}
	// position 是当前位置
	position int
	// mu 是保护位置的互斥锁
	mu sync.RWMutex
	// running 表示连接器是否正在运行
	running bool
}

// NewMemorySourceConnector 创建一个新的内存源连接器
func NewMemorySourceConnector() (core.SourceConnector, error) {
	return &MemorySourceConnector{}, nil
}

// Init 初始化连接器
func (c *MemorySourceConnector) Init(ctx context.Context, config map[string]interface{}) error {
	// 从配置中获取数据
	dataRaw, ok := config["data"]
	if !ok {
		return fmt.Errorf("data is required")
	}

	// 将数据转换为[]map[string]interface{}
	dataSlice, ok := dataRaw.([]map[string]interface{})
	if !ok {
		// 尝试从[]interface{}转换
		dataInterfaceSlice, ok := dataRaw.([]interface{})
		if !ok {
			return fmt.Errorf("data must be an array of objects")
		}

		// 将每个项目转换为map[string]interface{}
		dataSlice = make([]map[string]interface{}, len(dataInterfaceSlice))
		for i, item := range dataInterfaceSlice {
			itemMap, ok := item.(map[string]interface{})
			if !ok {
				return fmt.Errorf("data item %d is not an object", i)
			}
			dataSlice[i] = itemMap
		}
	}

	c.data = dataSlice
	c.position = 0
	return nil
}

// Start 启动连接器
func (c *MemorySourceConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.running = true
	return nil
}

// Stop 停止连接器
func (c *MemorySourceConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	c.running = false
	return nil
}

// Read 从源读取记录
func (c *MemorySourceConnector) Read(ctx context.Context) (*messages.ReadDataResponse, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// 检查是否正在运行
	if !c.running {
		return nil, fmt.Errorf("connector not running")
	}

	// 检查是否已读取所有数据
	if c.position >= len(c.data) {
		return &messages.ReadDataResponse{
			Success: true,
			Records: []messages.Record{},
			Offset:  c.position,
		}, nil
	}

	// 获取下一个项目
	item := c.data[c.position]
	c.position++

	// 将项目转换为JSON
	value, err := json.Marshal(item)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal item: %w", err)
	}

	// 创建记录
	record := messages.Record{
		Key:   []byte(fmt.Sprintf("%d", c.position-1)),
		Value: value,
	}

	return &messages.ReadDataResponse{
		Success: true,
		Records: []messages.Record{record},
		Offset:  c.position,
	}, nil
}

// Commit 提交偏移
func (c *MemorySourceConnector) Commit(ctx context.Context, offset interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// 检查是否正在运行
	if !c.running {
		return fmt.Errorf("connector not running")
	}

	// 将偏移转换为int
	offsetInt, ok := offset.(int)
	if !ok {
		return fmt.Errorf("offset must be an int")
	}

	// 更新位置
	c.position = offsetInt
	return nil
}

// 注册连接器
func init() {
	// 注册内存源连接器
	core.RegisterSourceConnector("memory", NewMemorySourceConnector)
}
